#include <boost/scoped_ptr.hpp>
#include <boost/scoped_array.hpp>
#include "node_environment/node_environment.h"
#include "Network.h"
#include "utils/logger.h"
#include "Op.h"
#include "worker.h"
#include "common.h"
#include "SerializableOutput.h"

using namespace internal;

void workerThread(const TaskList* tasks, int threadId) {
	network->setThreadId(threadId);

  int offset = threadId;
	for (int taskNumber = tasks->task_item_size() - offset - 1; taskNumber >= 0; taskNumber -= 2) {
		LOG("Worker %d, thread %d, task %d started.", network->my_node_number(), threadId, taskNumber);

		Op* taskOp = createTaskOp(tasks->task_item(taskNumber));
		taskOp->doit();

		LOG("Worker %d, thread %d, task %d finished.", network->my_node_number(), threadId, taskNumber);
	}
}

void consumeThread(const InputOperation* resultTypes, int threadId) {
	LOG("Sink thread started (as thread %d).", threadId);

	network->setThreadId(threadId);

	boost::scoped_ptr<DataSinkInterface> sink(network->openDataSink());

	int activeWorkers = network->nodes_count();
	LOG("%d active workers left.", activeWorkers);

	while (activeWorkers > 0) {
		std::size_t len;
		char flag;
		boost::scoped_array<char> receivedData(network->read(&len, &flag));
		CHECK(flag == CONSUME_MID_FLAG || flag == CONSUME_END_FLAG, "Expected consume message");

		LOG("Sink read %u bytes from network with %c flag.", len, flag);

		SerializableOutput data(*resultTypes);
		//LOG("SINK serializedLength %d, %d columns of type: %d %d", data.serializedLength(), resultTypes->type_size(), resultTypes->type(0), resultTypes->type(1)); // TODO remove it after debug
		data.deserialize(receivedData.get(), len);

		for (int i = 0; i < resultTypes->type_size(); ++i) {
			switch ((Type) resultTypes->type(i)) {
				case INT:
					sink->ConsumeInts(i, data.rowsCount(), (const int32_t*) data.column(i));
					break;

				case DOUBLE:
					sink->ConsumeDoubles(i, data.rowsCount(), (const double*) data.column(i));
					break;

				case BOOL:
					sink->ConsumeByteBools(i, data.rowsCount(), (const bool*) data.column(i));
					break;
			}
		}

		if (flag == CONSUME_END_FLAG) {
			--activeWorkers;
			LOG("Read end of stream message, %d active workers left.", activeWorkers);
		}
	}

	LOG("Sink thread finished.");
}
